package com;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.Test;

import java.time.Duration;
import java.util.*;

public class CustomConsumer2 {

    @Test
    public void test1() {
        // 从指定的topic消费消息，消费者有所属的group
        String topic = "order";
        String groupId = "group_demo3";

        // 创建配置对象
        Properties props = new Properties();

        // 指定集群地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node01:9092,node02:9092,node03:9092");


        // 消费者是按照offset偏移量来进行消费的，每条消息都有对应的offset，
        // offset按照从小到大消费，消费者会维护消费的offset，消费者默认按照上一次消费的offset继续消费
        // 自动提交模式值是true，提交的是offset
        // 消费者每次消费消息后，会把这条消息的offset自动提交kafka集群中
        // 会选择关闭自动提交，会带来重复消费的问题
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        /*
        auto.offset.reset：当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在，指定消费方式
        latest(默认) ：
            作为新的消费者组只消费自己启动之后发送到主题的消息
            非第一次启动消费者，会基于上次消费的offset继续消费
        earliest：第一次从头开始消费，以后按照消费offset记录继续消费

        什么时候使用 earliest，有一个需求，第一次需要基于所有的历史数据来做计算，之后重头消费一次
        每次都需要基于全量数据做计算，offset设置为0也可以达到从头消费的效果，始终都是重头消费
         */
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 设置消费者组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        // 指定key、value的反序列化
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 创建kafkaconsumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 消费者订阅指定的topic
        consumer.subscribe(Arrays.asList(topic));

        // 从头开始消费，必须要指定一个topic的分区
        // 消息回溯
        // consumer.assign(Arrays.asList(new TopicPartition(topic, 0)));
        // consumer.seekToBeginning(Arrays.asList(new TopicPartition(topic, 0)));

        // 持续性的拉取数据
        while (true) {
                // 消费者拉取消息, 在指定间隔时间后拉取一次，一次可能拉取多条数据，poll是以一个partition为单位来取数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
            for (ConsumerRecord record : records) {
                System.out.println("topic:" + record.topic() + "---partition:" + record.partition()
                        + "---offset:" + record.offset() + "---key:" + record.key()
                        + "---value:" + record.value());

            }
            consumer.commitSync();
        }

    }
}
